package org.example.producer;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

@Service
public class ProducerDemo {

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 启动zk：bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
     * 启动kafka：bin/kafka-server-start.sh -daemon config/server.properties
     * 创建topic：bin/kafka-topics.sh --create --zookeeper 172.17.0.5:2181 --replication-factor 1 --partitions 1 --topic lottery_invoice
     */
    public static final String TOPIC_INVOICE = "lottery_invoice";

    public void sendMessageInvoice(String message){

        ListenableFuture<SendResult<String, Object>> send = kafkaTemplate.send(TOPIC_INVOICE, message);
        send.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("MQ 消息发送失败，更新数据库表【等待定时任务扫码补偿MQ消息】");
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("MQ 消息发送完成，更新数据库表");
            }
        });

    }


}
